package net.anran.icecream;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.AbstractSelectableChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import net.anran.icecream.session.NioSession;
import net.anran.icecream.session.ReactorManager;

/**
 * Accept channel
 * 
 * @author guojje
 * 
 */
public class Reactor implements Runnable {

	private LinkedBlockingQueue<Object[]> register = new LinkedBlockingQueue<Object[]>();

	private ReactorManager reactorManager;

	private Selector selector;

	private TCPServerController tcpServerController;

	
	public Reactor(TCPServerController tcpServerController,
			ReactorManager reactorManager) {
		this.tcpServerController = tcpServerController;
		this.reactorManager = reactorManager;
	}

	public void start() throws IOException{
		selector = Selector.open();
		reactorManager.registerReactor(selector, this);
		new Thread(this).start();
	}
	
	public void registerSession(NioSession session, int eventType) {
		register.add(new Object[] { session, eventType });
	}

	public void registerChannel(AbstractSelectableChannel serverSocketChannel,
			int ops) {
		register.add(new Object[] { serverSocketChannel, ops });
	}

	
	private void processRegister() throws ClosedChannelException {
		Object[] o = register.poll();
		if (o != null) {
			int intestOp = ((Integer) o[1]).intValue();

			if (o[0] instanceof AbstractSelectableChannel) {
				AbstractSelectableChannel channel = (AbstractSelectableChannel) o[0];
				if (channel.isOpen()) {
					channel.register(selector, intestOp);
				}
				
			} else {

				NioSession session = (NioSession) o[0];
				SocketChannel sc = session.getSocketChannel();
				SelectionKey selectionKey = sc.keyFor(selector);
  
				if (selectionKey == null) {
					if (sc.isOpen()) {
						selectionKey = sc.register(selector, intestOp);
						session.setSelectionKey(selectionKey);
					}
				} else if (sc.isOpen() && selectionKey.isValid()) {
					selectionKey.interestOps(selectionKey.interestOps()
							| intestOp);
				}
			}
		}

	}

	@Override
	public void run() {
		try {
			while (selector.isOpen()) {

				processRegister();

				int keyNums = selector.select(1000);

				if (keyNums > 0) {
					Set<SelectionKey> selectedKeys = selector.selectedKeys();
					Iterator<SelectionKey> it = selectedKeys.iterator();

					while (it.hasNext()) {
						SelectionKey selectionKey = it.next();
						it.remove();
						dispatchEvent(selectionKey);
					}
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void dispatchEvent(SelectionKey selectionKey) {
		
		if (selectionKey.isValid()) {
			if (selectionKey.isAcceptable()) {
				try {
					tcpServerController.onAccept(selectionKey);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			if (selectionKey.isReadable()) {

				try {
					tcpServerController.onRead(selectionKey);
				} catch (ClosedChannelException e) {
					e.printStackTrace();
				}
			}
		}
	}
}
